package com.lengxf.flink.source;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.connector.source.util.ratelimit.RateLimiterStrategy;
import org.apache.flink.connector.datagen.source.DataGeneratorSource;
import org.apache.flink.connector.datagen.source.GeneratorFunction;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

//运行是增加参数--add-opens java.base/java.lang=ALL-UNNAMED --add-opens java.base/java.util=ALL-UNNAMED --add-opens java.base/java.nio=ALL-UNNAMED --add-opens java.base/sun.nio.ch=ALL-UNNAMED
public class DataGenerateDemo {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(2);
        DataGeneratorSource<String> source = new DataGeneratorSource<>(
                (GeneratorFunction<Long, String>) aLong -> "Number" + aLong,
                100,
                RateLimiterStrategy.perSecond(10),
                Types.STRING
        );

        env.fromSource(source, WatermarkStrategy.noWatermarks(), "dataGenerator")
                .print();


        env.execute();
    }
}
